package com.ct.ecommerce.base.listener;

import com.alibaba.fastjson.JSONObject;
import com.ct.ecommerce.api.base.client.DelayMessageService;
import com.ct.ecommerce.api.base.dto.DelayMessageDTO;
import com.ct.ecommerce.framework.constant.MqConstants;
import java.util.Date;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * @author changtong
 * @since 2021/6/29
 */
@Slf4j
@Component
@RocketMQMessageListener(
    topic = MqConstants.MQ_TOPIC_BASE,
    selectorExpression = MqConstants.MQ_TAG_BASE_DELAY_CREATE,
    consumerGroup = MqConstants.MQ_CONSUMER_GROUP_BASE
)
public class DelayMessageListener implements RocketMQListener<MessageExt> {

    @Resource
    private DelayMessageService delayMessageService;

    @Override
    public void onMessage(MessageExt message) {
        try {
            String data = new String(message.getBody());
            DelayMessageDTO delayMessage = JSONObject.parseObject(data, DelayMessageDTO.class);
            String times = message.getProperties().get(MqConstants.MQ_HEADER_BASE_TIMES);
            log.info("received times: {}, messageExt: {} ", times,
                JSONObject.toJSONString(message));
            delayMessageService.sendDelay(delayMessage, new Date(Long.parseLong(times)));
        }catch (Exception e){
            log.error("发送延迟消息失败: ",e);
        }
    }
}
